#  网站统计大数据
#  2017-12-27 20:00启动
#  参与人：壮志凌云

from __future__ import print_function
import re
from pyspark.sql import SparkSession
from pyspark.sql.functions import sum,row_number,desc,count,countDistinct,from_unixtime,max,isnull,instr,col,regexp_replace
from pyspark.sql import Window
import json

if __name__ == "__main__":
    spark = SparkSession \
        .builder \
        .appName("网站统计-访问分析") \
        .getOrCreate()

    #  采用python库函数，自行解析json，数据读取正常
    jsonData = spark.sparkContext.textFile("main_pv.json")
    WebData = jsonData.map(json.loads).toDF()
    WebData.printSchema()

    #  为数据添加日期、小时
    WebDataWithDH = WebData.withColumn('day', from_unixtime("vt", format='yyyy-MM-dd')).withColumn('hour',from_unixtime("vt",format='HH'))

    #  DataCube Create
    #  使用replace替换字符串，substring从字符串中提取子自付款，instr查找字符位置
    #  replace必须全字符匹配，更换为regexp_replace
    #  substr提取子字符串时，要求输入的起始位置和终止位置的数据类型相同。因此想办法创建一列，列的值为1
    WebDataVisit = WebDataWithDH.select(
        WebDataWithDH["day"],
        WebDataWithDH["hour"],
        WebDataWithDH["u"],
        WebDataWithDH["ip"],
        regexp_replace(regexp_replace(WebDataWithDH["url"],"http://",""),"https://","").alias('url'),
        WebDataWithDH["mi"]
#        substring("url",1,instr("url",'/')).alias("urlClass")
#    )
    ).withColumn("urlStop",instr('url','/') -1) \
    .withColumn("urlStart",col("urlStop")/col("urlStop")) \
    .withColumn('urlClass', col('url').substr(col('urlStart'),col('urlStop')))
#    .withColumn('urlClass',col('url').substr(1,col('urlAddr')-1))

    WebDataVisit.printSchema()
    WebDataVisit.show()

    #  网页获取参数
    #  考虑到参数需要更新，list参数换成字典参数
    AnalysisCondition = {'day':"day = '2017-12-26'", 'hour':'', "u":'', "ip": '', "ref":'', "refClass" : '', "mi" : '', "url" : ''}
    #  来源类型和来源网站选择参数
    urlAnalysis = ["url","urlClass"]

    WebDataVisitCube = WebDataVisit.cube(
        WebDataVisit["day"],
        WebDataVisit["hour"],
        WebDataVisit["u"],
        WebDataVisit["ip"],
        WebDataVisit["mi"],
        WebDataVisit["url"],
        WebDataVisit["urlClass"]
    ).count()

    WebDataVisitCube.show()

    ###################################################
    #  1.受访页面(指标概览、页面价值分析、入口页分析、退出页分析指标暂时不明白何区别，实现指标概览)
    ###################################################
    WebDataVisitedPage = WebDataVisitCube.filter(AnalysisCondition['day']) \
    .groupBy(urlAnalysis[0]) \
    .agg(
    max("count").alias("浏览量(PV)"),
    countDistinct("u").alias("访客数(UV)"),
    countDistinct("ip").alias("IP数")
    ).orderBy(desc("浏览量(PV)"))

    WebDataVisitedPage.show()

    ###################################################
    #  2.入口域名(什么叫做入口域名，不明白定义，暂时搁置)
    ###################################################

    ###################################################
    #  3.受访域名(什么叫做受访域名，不明白定义，暂时认为就是形如 www.baike.com 这样的url的网址一级域名)
    ###################################################
    WebDataVisitedDomain = WebDataVisitCube.filter(AnalysisCondition['day']) \
    .groupBy(urlAnalysis[1]) \
    .agg(
    max("count").alias("浏览量(PV)"),
    countDistinct("u").alias("访客数(UV)"),
    countDistinct("ip").alias("IP数")
    ).orderBy(desc("浏览量(PV)"))

    WebDataVisitedDomain.show()

    ###################################################
    #  4.页面点击图(现有数据暂时不支持此分析)
    ###################################################

    ###################################################
    #  5.页面上下游(现有数据暂时不支持此分析)
    ###################################################

    ###################################################
    #  6.事件分析(现有数据暂时不支持此分析)
    ###################################################

    ###################################################
    #  7.子目录(百度统计无法刷新Flash，不知是何统计)
    ###################################################

    spark.stop()
